package org.gg.boom.kernel.web;

import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.gg.boom.common.utils.ServletUtils;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;


@Slf4j
@Component
public class SseServer {
    /**
     * msg池
     */
    private final Map<String, SseEmitter> currConnPool = new ConcurrentHashMap<>();
    /**
     * 心跳检测池
     */
    private final Map<String, Integer> alivePool = new ConcurrentHashMap<>();

    /**
     * 创建连接
     *
     * @param emitterKey
     * @return
     */
    public SseEmitter createConn(String emitterKey) {
        // 完成已存在的Emitter连接
        if (currConnPool.containsKey(emitterKey)) {
            currConnPool.get(emitterKey).complete();
        }

        // 创建SSE
        SseEmitter emitter = new SseEmitter(0L);
        // 断开回调
        emitter.onCompletion(() -> {
            log.info("emitterKey:{}，断开连接", emitterKey);
        });
        // 超时回调
        emitter.onTimeout(() -> {
            log.info("emitterKey:{}，超时断开连接", emitterKey);
            removeConn(emitterKey);
        });
        // 报错回调
        emitter.onError((e) -> {
            log.info("emitterKey:{}，连接异常断开连接", emitterKey);
            removeConn(emitterKey);
        });
        currConnPool.put(emitterKey, emitter);
        alivePool.put(emitterKey, 60);
        // 设置响应头，防止Nginx缓冲
        ServletUtils.getResponse().setHeader("X-Accel-Buffering", "no");
        return emitter;
    }

    /**
     * 断开连接
     * 1、完成Emitter
     * 2、移除Emitter连接池
     *
     * @param emitterKey
     */
    public void removeConn(String emitterKey) {
        if (currConnPool.containsKey(emitterKey)) {
            complete(emitterKey);
            currConnPool.remove(emitterKey);
            alivePool.remove(emitterKey);
        }
    }

    /**
     * 完成Emitter推送
     * 1、如果有且只有一个连接那么也会把任务关掉
     * 2、将Emitter置为完成状态
     */
    private void complete(String emitterKey) {
        currConnPool.get(emitterKey).complete();
    }


    /**
     * 指定Emitter 发送一次消息
     *
     * @param emitterKey
     * @param supplier
     */
    public void sendMsgOnce(String emitterKey, Supplier<Object> supplier) {
        if (currConnPool.containsKey(emitterKey)) {
            try {
                currConnPool.get(emitterKey).send(supplier.get());
            } catch (IllegalStateException e) {
                log.error("发送消息失败【连接状态变更】 -->emitterKey:{}, error:{}", emitterKey, e.getMessage());
                removeConn(emitterKey);
            } catch (IOException e) {
                log.error("发送消息失败【IO异常】 -->emitterKey:{}, error:{}", emitterKey, e.getMessage());
                removeConn(emitterKey);
            } catch (Exception e) {
                log.error("发送消息失败【其他异常】 -->emitterKey:{}, error:{}", emitterKey, e.getMessage());
            }
        }
    }

    public void refreshAliveTime(String emitterKey) {
        alivePool.computeIfPresent(emitterKey, (k, v) -> 60);
    }

    @Scheduled(fixedDelay = 20000)
    public void clearConnPool() {
        alivePool.forEach((k, v) -> {
            if (v <= 0) {
                removeConn(k);
            } else {
                alivePool.put(k, v - 20);
            }
        });
    }

    @PreDestroy
    public void destroy() {
        currConnPool.clear();
    }
}
